package helloworld;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;

/**
 * @author ChenJunhong
 * @date 2018/12/19
 * Description:
 */
public class Main {
    public static void main(String[] args) throws InterruptedException {

        // 构造ringbuffer
        RingBuffer<Order> ringBuffer = RingBuffer.create(
                ProducerType.MULTI,
                new OrderFactory(),
                1024*1024,
                new BlockingWaitStrategy()
        );

        // 创建一个序号屏障
        SequenceBarrier barrier = ringBuffer.newBarrier();

        // 创建消费者数组
        Comsumer[] comsumers = new Comsumer[10];
        for (int i = 0; i < comsumers.length; i++) {
            comsumers[i] = new Comsumer("comsumer"+i);
        }

        // 创建多消费者的工作池
        WorkerPool<Order> workerPool = new WorkerPool<Order>(
                ringBuffer,
                barrier,
                new EventExceptionHandler(),
                comsumers
        );

        // 设置多个消费者的sequence序号 用于单独统计消费进度, 并且设置到ringbuffer中
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());

        workerPool.start(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()));

        // -----消费者创建完毕-----


        // -----创建生产者------
        // 用于阻塞生产者生产数据
        CountDownLatch latch = new CountDownLatch(1);

        for(int i = 0; i < 100; i++) {
            final Producer producer = new Producer(ringBuffer);
            new Thread(new Runnable() {
                public void run() {
                    try {
                        // 阻塞生产者
                        latch.await();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    for(int j = 0; j<100; j++) {
                        producer.sendData(UUID.randomUUID().toString());
                    }
                }
            }).start();
        }

        System.err.println("----------线程创建完毕，开始生产数据----------");
        // 放行生产者，让其生产数据
        latch.countDown();

        // 等待消费者消费完数据
        Thread.sleep(2000);

        System.err.println("消费次数："+Comsumer.count);

    }

    static class EventExceptionHandler implements ExceptionHandler<Order> {
        public void handleEventException(Throwable ex, long sequence, Order event) {
        }

        public void handleOnStartException(Throwable ex) {
        }

        public void handleOnShutdownException(Throwable ex) {
        }
    }
}
